package com.dd.ms.log.web.service.internal;

import com.dd.ms.log.api.pojo.dto.LogEntryDTO;
import com.dd.ms.log.api.pojo.enumuration.LogLevelEnum;
import com.dd.ms.log.web.converter.LogEntryConverter;
import com.dd.ms.log.web.pojo.LogEntryESDO;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @Author liuxianmeng
 * @CreateTime 2025/6/28 14:44
 * @Description 类功能作用说明
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class LogConsumer {

    private final ElasticService elasticService;

    // 关键注解：多个线程并发消费同一主题
    @KafkaListener(topics = "${spring.kafka.topic.system_log}", groupId = "${spring.kafka.consumer.group-id}")
    public void consumeLogs(ConsumerRecord<String, Object> record, Acknowledgment acknowledgment) {
        try {
            // 1. 解析原始日志
            LogEntryDTO logEntryDTO = (LogEntryDTO) record.value();

            // 2. 处理耗时日志（响应时间>1s）
            if (logEntryDTO.getDurationMs() != null && logEntryDTO.getDurationMs() > 1000) {
                log.warn("慢响应（耗时日志）记录: {}ms in {}",
                        logEntryDTO.getDurationMs(), logEntryDTO.getServiceName());
            }

            // 3. 存储到Elasticsearch
            LogEntryESDO logEntryESDO = LogEntryConverter.logEntryBO2ESDO(logEntryDTO);
            elasticService.add(logEntryESDO);

            // 4. 错误日志报警
            if (LogLevelEnum.ERROR.name().equals(logEntryDTO.getLevel())) {
                // 发送告警通知
                log.error("Error logged: {}", logEntryDTO.getMessage());
            }

            // 手动提交偏移量
            acknowledgment.acknowledge();
        } catch (Exception e) {
            log.error("Failed to process log: {}", record.value(), e);
        }
    }
}